package io.kiki.sba.registry.client.provider;


import io.kiki.sba.registry.client.api.Client;
import io.kiki.sba.registry.client.api.*;
import io.kiki.sba.registry.client.api.exception.DuplicateException;
import io.kiki.sba.registry.client.api.exception.RegistryClientException;
import io.kiki.sba.registry.client.api.model.RegistryType;
import io.kiki.sba.registry.client.api.registration.ConfiguratorRegistration;
import io.kiki.sba.registry.client.api.registration.PublisherRegistration;
import io.kiki.sba.registry.client.api.registration.SubscriberRegistration;
import io.kiki.sba.registry.client.auth.AuthManager;
import io.kiki.sba.registry.client.auth.NoopAuthManager;
import io.kiki.sba.registry.client.event.ConfiguratorProcessEvent;
import io.kiki.sba.registry.client.event.EventBusImpl;
import io.kiki.sba.registry.client.event.LookoutSubscriber;
import io.kiki.sba.registry.client.event.SubscriberProcessEvent;
import io.kiki.sba.registry.client.remoting.*;
import io.kiki.sba.registry.client.task.ObserverHandler;
import io.kiki.sba.registry.client.task.SyncConfigThread;
import io.kiki.sba.registry.client.task.TaskEvent;
import io.kiki.sba.registry.client.task.WorkerThread;
import io.kiki.sba.registry.client.util.StringUtils;
import io.kiki.sba.registry.core.model.ReceivedConfigData;
import io.kiki.sba.registry.core.model.ReceivedData;
import io.kiki.stack.netty.channelManager.ChannelEventProcessor;
import io.kiki.stack.netty.channelManager.ChannelEventType;
import io.kiki.stack.netty.userProcessor.UserProcessor;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;

import static io.kiki.sba.registry.client.constants.ValueConstants.DEFAULT_GROUP;

@Setter
@Getter
public class ClientImpl implements Client {


    private static final Logger logger = LoggerFactory.getLogger(ClientImpl.class);

    private ClientConfig clientConfig;

    private RegisterCache registerCache;

    private ServerManager serverManager;

    private WorkerThread workerThread;

    private ClientChannel clientChannel;


    private Map<Class<?>, UserProcessor> userProcessorMap;


    private Map<ChannelEventType, ChannelEventProcessor> channelEventTypeToChannelEventProcessorMap;

    private ConcurrentMap<PublisherRegistration, Publisher> registrationPublisherMap;

    private ConcurrentMap<SubscriberRegistration, Subscriber> registrationSubscriberMap;

    private ConcurrentMap<ConfiguratorRegistration, Configurator> registrationConfiguratorMap;


    @Setter
    private ObserverHandler observerHandler;

    @Setter
    private AuthManager authManager;


    @Setter
    private EventBus eventBus;


    @Setter
    private LookoutSubscriber lookoutSubscriber;

    private AtomicBoolean init = new AtomicBoolean(false);

    public ClientImpl(ClientConfig clientConfig) {
        // clone config to avoid configuration changes at runtime
        this.clientConfig = cloneConfig(clientConfig);
        this.registerCache = new RegisterCache();
        this.registrationPublisherMap = new ConcurrentHashMap<PublisherRegistration, Publisher>();
        this.registrationSubscriberMap = new ConcurrentHashMap<SubscriberRegistration, Subscriber>();
        this.registrationConfiguratorMap = new ConcurrentHashMap<ConfiguratorRegistration, Configurator>();
    }

    private DefaultClientConfig cloneConfig(ClientConfig clientConfig) {
        DefaultClientConfig cloneConfig = null;
        DefaultRegistryClientConfigBuilder builder = DefaultRegistryClientConfigBuilder.start();
        if (null != clientConfig) {
            cloneConfig = builder.setEnv(clientConfig.getEnv()).setAppName(clientConfig.getAppName()).setInstanceId(clientConfig.getInstanceId()).setDataCenter(clientConfig.getDataCenter()).setZone(clientConfig.getZone()).setRegistryEndpoint(clientConfig.getRegistryEndpoint()).setRegistryEndpointPort(clientConfig.getRegistryEndpointPort()).setConnectTimeout(clientConfig.getConnectTimeout()).setSocketTimeout(clientConfig.getSocketTimeout()).setInvokeTimeout(clientConfig.getInvokeTimeout()).setRecheckInterval(clientConfig.getRecheckInterval()).setObserverThreadCoreSize(clientConfig.getObserverThreadCoreSize()).setObserverThreadMaxSize(clientConfig.getObserverThreadMaxSize()).setObserverThreadQueueLength(clientConfig.getObserverThreadQueueLength()).setObserverCallbackTimeout(clientConfig.getObserverCallbackTimeout()).setSyncConfigRetryInterval(clientConfig.getSyncConfigRetryInterval()).setAccessKey(clientConfig.getAccessKey()).setSecretKey(clientConfig.getSecretKey()).build();
        }
        return cloneConfig;
    }

    public void init() {
        if (!init.compareAndSet(false, true)) {
            return;
        }

        // init lookout subscriber
        if (null == lookoutSubscriber) {
            this.lookoutSubscriber = new LookoutSubscriber();
        }

        // init event bus
        if (null == eventBus) {
            this.eventBus = new EventBusImpl(clientConfig);
            this.eventBus.register(SubscriberProcessEvent.class, lookoutSubscriber);
            this.eventBus.register(ConfiguratorProcessEvent.class, lookoutSubscriber);
        }

        // init server manager
        if (null == serverManager) {
            this.serverManager = new ServerManagerImpl(clientConfig);
        }

        // init observer handler
        if (null == observerHandler) {
            observerHandler = new DefaultObserverHandler(clientConfig, eventBus);
        }

        // init auth manager
        if (null == authManager) {
            authManager = NoopAuthManager.INSTANCE;
        }

        // init user processor
        List<UserProcessor> userProcessorList = new ArrayList<UserProcessor>();
        if (null == userProcessorMap) {
            userProcessorList.add(new ReceivedDataProcessor(registerCache, observerHandler));
            userProcessorList.add(new ReceivedConfigDataProcessor(registerCache, observerHandler));
        } else {
            UserProcessor userProcessor = userProcessorMap.get(ReceivedData.class);
            if (null == userProcessor) {
                userProcessorList.add(new ReceivedDataProcessor(registerCache, observerHandler));
            }
            userProcessor = userProcessorMap.get(ReceivedConfigData.class);
            if (null == userProcessor) {
                userProcessorList.add(new ReceivedConfigDataProcessor(registerCache, observerHandler));
            }
            userProcessorList.addAll(userProcessorMap.values());
        }

        // init connection event processor
        if (null == channelEventTypeToChannelEventProcessorMap) {
            channelEventTypeToChannelEventProcessorMap = new HashMap<ChannelEventType, ChannelEventProcessor>(ChannelEventType.values().length);
        }
        if (null == channelEventTypeToChannelEventProcessorMap.get(ChannelEventType.CLOSE)) {
            ClientConnectionCloseEventProcessor connectionCloseEventProcessor = new ClientConnectionCloseEventProcessor();
            channelEventTypeToChannelEventProcessorMap.put(ChannelEventType.CLOSE, connectionCloseEventProcessor);
        }
        if (null == channelEventTypeToChannelEventProcessorMap.get(ChannelEventType.CONNECT)) {
            ClientConnectionOpenEventProcessor connectionOpenEventProcessor = new ClientConnectionOpenEventProcessor();
            channelEventTypeToChannelEventProcessorMap.put(ChannelEventType.CONNECT, connectionOpenEventProcessor);
        }

        // init client connection and register worker
        clientChannel = new ClientChannel(serverManager, userProcessorList, channelEventTypeToChannelEventProcessorMap, registerCache, clientConfig);

        workerThread = new WorkerThread(clientChannel, clientConfig, registerCache);
        clientChannel.setWorker(workerThread);

        clientChannel.init();

        // init registry check thread
        new RegistryCheckThread().start();

        // init sync config thread
        new SyncConfigThread(clientChannel, registerCache, clientConfig, observerHandler).start();
    }


    @Override
    public Publisher register(PublisherRegistration publisherRegistration, String... data) {
        if (!init.get()) {
            throw new IllegalStateException("Client needs to be initialized before using.");
        }

        if (null == publisherRegistration) {
            throw new IllegalArgumentException("Registration can not be null.");
        }

        if (StringUtils.isBlank(publisherRegistration.getDataId())) {
            throw new IllegalArgumentException("DataId can not be null");
        }

        if (StringUtils.isBlank(publisherRegistration.getGroup())) {
            publisherRegistration.setGroup(DEFAULT_GROUP);
        }

        Publisher publisher = registrationPublisherMap.get(publisherRegistration);

        if (null != publisher) {
            throwDuplicateException(publisherRegistration, publisher);
        }

        publisher = new PublisherImpl(publisherRegistration, workerThread, clientConfig);
        ((PublisherImpl) publisher).setAuthManager(authManager);

        Publisher oldPublisher = registrationPublisherMap.putIfAbsent(publisherRegistration, publisher);
        if (null != oldPublisher) {
            throwDuplicateException(publisherRegistration, oldPublisher);
        }

        registerCache.addRegister(publisher);

        publisher.republish(data);

        logger.info("[api] Regist publisher success, dataId: {}, group: {}, registerId: {}", publisher.getDataId(), publisher.getGroup(), publisher.getRegistId());

        return publisher;
    }

    private void throwDuplicateException(PublisherRegistration registration, Publisher publisher) {
        logger.info("[api] Publisher already exists, dataId: {}, group: {}, registerId: {}", publisher.getDataId(), publisher.getGroup(), publisher.getRegistId());

        throw new DuplicateException("Duplicate Publisher registration. (dataId: " + registration.getDataId() + ", group: " + registration.getGroup() + ")");
    }


    @Override
    public Subscriber register(SubscriberRegistration subscriberRegistration) {
        if (!init.get()) {
            throw new IllegalStateException("Client needs to be initialized before using.");
        }

        if (null == subscriberRegistration) {
            throw new IllegalArgumentException("Registration can not be null.");
        }

        if (StringUtils.isBlank(subscriberRegistration.getDataId())) {
            throw new IllegalArgumentException("DataId can not be null.");
        }

        if (null == subscriberRegistration.getSubscriberDataObserver()) {
            throw new IllegalArgumentException("Subscriber data observer can not be null.");
        }

        if (StringUtils.isBlank(subscriberRegistration.getGroup())) {
            subscriberRegistration.setGroup(DEFAULT_GROUP);
        }

        Subscriber subscriber = registrationSubscriberMap.get(subscriberRegistration);

        if (null != subscriber) {
            throwDuplicateException(subscriberRegistration, subscriber);
        }

        subscriber = new SubscriberImpl(subscriberRegistration, workerThread, clientConfig);
        ((SubscriberImpl) subscriber).setAuthManager(authManager);

        Subscriber oldSubscriber = registrationSubscriberMap.putIfAbsent(subscriberRegistration, subscriber);
        if (null != oldSubscriber) {
            throwDuplicateException(subscriberRegistration, oldSubscriber);
        }

        registerCache.addRegister(subscriber);
        addRegisterTask(subscriber);

        logger.info("[api] Regist subscriber success, dataId: {}, group: {}, scope: {}, registerId: {}", subscriber.getDataId(), subscriber.getGroup(), subscriber.getScopeEnum(), subscriber.getRegistId());

        return subscriber;
    }


    @Override
    public Configurator register(ConfiguratorRegistration configuratorRegistration) {
        if (!init.get()) {
            throw new IllegalStateException("Client needs to be initialized before using.");
        }

        if (null == configuratorRegistration) {
            throw new IllegalArgumentException("Registration can not be null.");
        }

        if (StringUtils.isBlank(configuratorRegistration.getDataId())) {
            throw new IllegalArgumentException("DataId can not be null.");
        }

        if (null == configuratorRegistration.getConfigDataObserver()) {
            throw new IllegalArgumentException("Config data observer can not be null");
        }

        if (StringUtils.isBlank(configuratorRegistration.getGroup())) {
            configuratorRegistration.setGroup(DEFAULT_GROUP);
        }

        Configurator configurator = registrationConfiguratorMap.get(configuratorRegistration);

        if (null != configurator) {
            throwDuplicateException(configurator);
        }

        configurator = new ConfiguratorImpl(configuratorRegistration, clientConfig, workerThread);
        ((ConfiguratorImpl) configurator).setAuthManager(authManager);

        Configurator oldConfigurator = registrationConfiguratorMap.putIfAbsent(configuratorRegistration, configurator);
        if (null != oldConfigurator) {
            throwDuplicateException(configurator);
        }

        registerCache.addRegister(configurator);
        addRegisterTask(configurator);

        logger.info("[api] Regist configurator success, dataId: {}, registerId: {}", configurator.getDataId(), configurator.getRegistId());

        return configurator;
    }

    private void throwDuplicateException(Configurator configurator) {
        logger.info("[api] Configurator already exists, dataId: {}, registerId: {}", configurator.getDataObserver(), configurator.getRegistId());
        throw new DuplicateException("Duplicate configurator registration. (dataId: " + configurator.getDataId() + " )");
    }


    @Override
    public int unregister(String dataId, String group, RegistryType registryType) {
        if (StringUtils.isBlank(dataId)) {
            throw new IllegalArgumentException("dataId can not be empty");
        }

        if (null == registryType) {
            throw new IllegalArgumentException("registry type can not be null");
        }

        if (null == group) {
            group = DEFAULT_GROUP;
        }

        List<Register> registers = new ArrayList<Register>();

        if (RegistryType.publisher == registryType) {
            List<Publisher> publishers = registerCache.getPublisherList();
            for (Publisher publisher : publishers) {
                if (dataId.equals(publisher.getDataId()) && group.equals(publisher.getGroup())) {
                    registers.add(publisher);
                }
            }
        } else
            if (RegistryType.subscriber == registryType) {
                List<Subscriber> subscriberList = registerCache.getSubscriberList();
                for (Subscriber subscriber : subscriberList) {
                    if (dataId.equals(subscriber.getDataId()) && group.equals(subscriber.getGroup())) {
                        registers.add(subscriber);
                    }
                }
            } else
                if (RegistryType.configurator == registryType) {
                    List<Configurator> configuratorList = registerCache.getConfiguratorList();
                    for (Configurator configurator : configuratorList) {
                        if (dataId.equals(configurator.getDataId())) {
                            registers.add(configurator);
                        }
                    }
                }

        for (Register register : registers) {
            register.unregister();
        }
        return registers.size();
    }

    private void throwDuplicateException(SubscriberRegistration registration, Subscriber subscriber) {
        logger.info("[api] Subscriber already exists, dataId: {}, group: {}, scope: {}, registerId: {}", subscriber.getDataId(), subscriber.getGroup(), subscriber.getScopeEnum(), subscriber.getRegistId());
        throw new DuplicateException("Duplicate subscriber registration. (dataId: " + registration.getDataId() + ", group: " + registration.getGroup() + ")");
    }


    private void addRegisterTask(Register register) throws RegistryClientException {
        try {
            TaskEvent event = new TaskEvent(register);
            workerThread.schedule(event);
        } catch (Exception e) {
            logger.error("Register task schedule error, {}", register, e);
            throw new RegistryClientException("Register task schedule error", e);
        }
    }


    class RegistryCheckThread extends Thread {


        public RegistryCheckThread() {
            super("RegistryClientCheckThread");
            setDaemon(true);
        }


        @Override
        public void run() {

            //noinspection InfiniteLoopStatement
            while (true) {
                try {
                    Thread.sleep(clientConfig.getRecheckInterval());

                    List<Publisher> allPublishers = registerCache.getPublisherList();

                    for (Publisher publisher : allPublishers) {
                        try {
                            if (null != publisher && !((AbstractInternalRegister) publisher).isDone()) {
                                addRegisterTask(publisher);
                            }
                        } catch (Exception e) {
                            logger.error("Sync publisher error, {}", publisher, e);
                        }
                    }

                    List<Subscriber> subscriberList = registerCache.getSubscriberList();

                    for (Subscriber subscriber : subscriberList) {
                        try {
                            if (null != subscriber && !((AbstractInternalRegister) subscriber).isDone()) {
                                addRegisterTask(subscriber);
                            }
                        } catch (Exception e) {
                            logger.error("Sync subscriber error, {}", subscriber, e);
                        }
                    }

                    List<Configurator> configuratorList = registerCache.getConfiguratorList();

                    for (Configurator configurator : configuratorList) {
                        try {
                            if (null != configurator && !((AbstractInternalRegister) configurator).isDone()) {
                                addRegisterTask(configurator);
                            }
                        } catch (Exception e) {
                            logger.error("Sync configurator error, {}", configurator, e);
                        }
                    }
                } catch (Throwable e) {
                    logger.error("Execute error", e);
                }
            }
        }
    }
}
